package de.mmis.core.infrastructure;

import static de.mmis.core.base.infrastructure.InfrastructureEvent.InfrastructureEventType.*;
import static de.mmis.core.base.infrastructure.InfrastructureException.Type.*;
import static de.mmis.core.config.Config.*;
import static de.mmis.core.infrastructure.HostServer.*;
import static de.mmis.core.infrastructure.Message.MessageType.*;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import javax.xml.ws.Holder;

import org.apache.log4j.Logger;

import de.mmis.core.base.Pair;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.abstracttree.TreeParserException;
import de.mmis.core.base.abstracttree.serializer.DeserializableException;
import de.mmis.core.base.event.Observer;
import de.mmis.core.base.infrastructure.Infrastructure;
import de.mmis.core.base.infrastructure.InfrastructureEvent;
import de.mmis.core.base.infrastructure.InfrastructureEvent.InfrastructureEventType;
import de.mmis.core.base.infrastructure.InfrastructureException;
import de.mmis.core.base.infrastructure.PublishedDeviceDescription;
import de.mmis.core.sexpression.SExpressionParser;
import de.mmis.core.sexpression.SExpressionWriter;

/**
 * <p>
 * The infrastructure communicates with the {@link HostServer} and creates a
 * {@link HostServer} if they don't exist on the local machine. It also manages
 * a list of VM-local devices, so they can be restored in case the VM with the
 * Server crashes.
 * </p>
 * 
 * <p>
 * There may only be one instance of this class in one VM. If you want to
 * destroy the Infrastructure of the current VM, you have to call
 * {@link #stop()} explicitly.
 * </p>
 * 
 * @author Martin Nyolt
 * 
 */
public class DistributedInfrastructure implements Infrastructure {

	private final static Logger LOGGER = Logger
			.getLogger(DistributedInfrastructure.class);

	class InfrastructureThread extends Thread {

		public InfrastructureThread() throws InfrastructureException {
			super(INFRASTRUCTURE_THREAD_GROUP, "InfrastructureThread");
			setDaemon(true);

			receiveHelloMessage();
		}

		private void receiveHelloMessage() throws InfrastructureException {
			try {
				Tree h = SExpressionParser.parse(input, false);
				Message message = h.getAs(Message.class);
				if (message.getType() != HELLO) {
					LOGGER.error("ignore " + message);

					h = SExpressionParser.parse(input, false);
					message = h.getAs(Message.class);

					if (message.getType() != HELLO)
						throw new InfrastructureException(
								"Message is not a hello message: " + message);
				}
				bindAddress = message.getAddress();
			} catch (TreeParserException e) {
				throw new InfrastructureException(
						"Error receiving hello message", e);
			} catch (IOException e) {
				throw new InfrastructureException(
						"Error receiving hello message", e);
			} catch (DeserializableException e) {
				throw new InfrastructureException(
						"Error receiving hello message", e);
			}

			LOGGER.debug("saved bindAddress: " + bindAddress);
		}

		@Override
		public void run() {
			// if an exception caused that we are no longer connected to the
			// infrastructure, it's saved here
			Exception shutDownException = null;

			// add all existing devices
			synchronized (devices) {
				for (PublishedDeviceDescription device : devices) {
					Message message = Message.addDevice(device, false);
					if (sendMessage(message) == false) {
						String m = "New socket was unable to send - give up";
						LOGGER.fatal(m);
						shutDownException = new Exception(m);
						running = false;
						break;
					}
				}
			}

			while (running) {
				while (!clientSocket.isClosed()) {
					Tree expr = null;
					try {
						expr = SExpressionParser.parse(input, false);
					} catch (TreeParserException e) {
						if (running && !e.unexpectedEndOfStreamOccurred())
							LOGGER.error(
									"Can't parse S-Expression from socket", e);
						break;
					} catch (IOException e) {
						if (running)
							LOGGER.error("Can't read from socket", e);
						break;
					}

					if (expr == null)
						continue;

					try {
						Message message = expr.getAs(Message.class);
						parseMessage(message);
					} catch (DeserializableException e) {
						LOGGER.error("Could not deserialize message " + expr, e);
					}
				}

				clientSocket = null;
				input = null;
				output = null;

				if (!running)
					break;

				// if the old server accounced it wants to stop
				boolean clean = false;
				long diff = System.currentTimeMillis() - shutDownAnnounceTime;
				if (diff < 60 * 1000)
					// clean if the announcement was at most 1 minute ago
					clean = true;

				LOGGER.info("creating new sockets");

				// try to create new server and connect to it
				try {
					setUpServer(false, clean);
					receiveHelloMessage();
				} catch (InfrastructureException e) {
					LOGGER.fatal(
							"Can't create sockets or connect to infrastructure",
							e);
					running = false;
					shutDownException = e;
					break;
				}

				// add all existing devices
				synchronized (devices) {
					for (PublishedDeviceDescription device : devices) {
						Message message = Message.addDevice(device, true);
						if (sendMessage(message) == false) {
							String m = "New socket was unable to send - give up";
							LOGGER.fatal(m);
							shutDownException = new Exception(m);
							running = false;
							break;
						}
					}
				}

				// notify that a new VM on this machine manages the devices
				// other machines get this message through the RemoteServer
				InfrastructureEvent event = new InfrastructureEvent(
						InfrastructureEventType.MIGRATED);
				event.address = clientSocket.getLocalAddress();
				event.clean = clean;
				notifyObservers(event);
			} // while (running)

			// notify observers we are no longer connected to
			// the Infrastructure
			InfrastructureEvent event = new InfrastructureEvent(
					InfrastructureEventType.SHUT_DOWN);
			event.exception = shutDownException;
			event.clean = shutDownException == null;
			notifyObservers(event);
		}

	}

	private static DistributedInfrastructure instance;
	private final static Object instanceMonitor = new Object();

	private final LinkedList<Observer<InfrastructureEvent>> observers = new LinkedList<Observer<InfrastructureEvent>>();
	private final LinkedList<PublishedDeviceDescription> devices = new LinkedList<PublishedDeviceDescription>();

	private Socket clientSocket;
	private BufferedInputStream input;
	private BufferedOutputStream output;

	private boolean running;
	private final Holder<Boolean> waitForInfrastructureSetup = new Holder<Boolean>();

	private InetAddress bindAddress;
	private HostServer hostServer = null;

	/**
	 * Stores all waiting single device searches. {@link Object#notifyAll()} is
	 * called, when an answer matches the pattern. The pattern is then replaces
	 * with the result. The method comaring an "answer" event with the patterns
	 * have to remove the entry once it matched.
	 */
	private LinkedList<Pair<PublishedDeviceDescription, Object>> searches = new LinkedList<Pair<PublishedDeviceDescription, Object>>();

	/**
	 * The time when the local infrastructure server announces it wants to shut
	 * down.
	 */
	private long shutDownAnnounceTime;

	private DistributedInfrastructure(InetAddress bindAddress, boolean start)
			throws InfrastructureException {
		Thread shutdownHook = new Thread(INFRASTRUCTURE_THREAD_GROUP,
				new Runnable() {
					@Override
					public void run() {
						DistributedInfrastructure.this.stop();
					}
				}, "shutdown hook");

		Runtime.getRuntime().addShutdownHook(shutdownHook);

		if (start)
			start(bindAddress);
	}

	public void start(InetAddress bindAddress) throws InfrastructureException {
		if (running)
			throw new InfrastructureException(
					"Infrastructure is already running");

		this.bindAddress = bindAddress;
		running = true;
		setUpServer(true, true);
		new InfrastructureThread().start();
	}

	public static List<InetAddress> getPossibleBindAddresses() {
		return HostServer.getPossibleBindAddresses();
	}

	public void rebind(InetAddress bindAddress) throws InfrastructureException {
		this.bindAddress = bindAddress;
		sendMessage(Message.rebind(bindAddress));
		stop();
		start(bindAddress);
	}

	public InetAddress getBindAddress() {
		return bindAddress;
	}

	private Socket createSocket() {
		Socket result = null;
		input = null;
		output = null;
		try {
			result = new Socket((String) null, INFRASTRUCTURE_PORT);
			input = new BufferedInputStream(result.getInputStream());
			output = new BufferedOutputStream(result.getOutputStream());
		} catch (UnknownHostException e) {
			// should not happen
			LOGGER.error("could not find localhost", e);
		} catch (IOException e) {
			// just return null
		}

		return result;
	}

	/**
	 * blocks until the infrastructure has been set up
	 * 
	 * @throws InfrastructureException
	 *             if the Infrastructure is not running
	 */
	private void waitForInfrastructureSetup() throws InfrastructureException {
		if (!running)
			throw new InfrastructureException("Infrastructure is not running")
					.setType(NOT_RUNNING);

		synchronized (waitForInfrastructureSetup) {
			while (waitForInfrastructureSetup.value.booleanValue())
				try {
					waitForInfrastructureSetup.wait();
				} catch (InterruptedException e) {
					// ignore
				}
		}

		if (!running)
			throw new InfrastructureException("Infrastructure is not running")
					.setType(NOT_RUNNING);
	}

	/**
	 * creates the {@link LocalServer} and {@link HostServer}, if they don't
	 * exist, and connects to the {@link LocalServer}
	 * 
	 * @param first
	 *            true if this is a new machine, false if it's taking over
	 *            another infrastructure of this machine
	 * @param clean
	 *            false if the old VM did not announced it wants to stop
	 * 
	 * @throws InfrastructureException
	 *             when no connection could be made
	 */
	private void setUpServer(boolean first, boolean clean)
			throws InfrastructureException {
		// don't do anything until the infrastructure is set up
		synchronized (waitForInfrastructureSetup) {
			waitForInfrastructureSetup.value = Boolean.TRUE;
		}

		clientSocket = createSocket();

		// fast forward if everything is fine
		if (clientSocket != null) {
			synchronized (waitForInfrastructureSetup) {
				waitForInfrastructureSetup.value = Boolean.FALSE;
				waitForInfrastructureSetup.notifyAll();
			}
			return;
		}

		InfrastructureException exception = null;
		try {
			InfrastructureException serverException = null;
			// there is no server running - create one
			try {
				if (hostServer == null)
					hostServer = new HostServer();
				try {
					hostServer.start(first, clean, bindAddress);
				} catch (Exception e) {
					throw new InfrastructureException(
							"Could not start remote Server", e);
				}
			} catch (InfrastructureException e) {
				// unable to create local servers, mostly because
				// we were too slow, so there should now be another server

				// save exception
				serverException = e;
			} finally {
				// wait so the server should be ready to accept connections
				try {
					Thread.sleep(250);
				} catch (InterruptedException e) {
					// ignore
				}

				// connect to local server
				clientSocket = createSocket();

				if (clientSocket == null) {
					synchronized (waitForInfrastructureSetup) {
						waitForInfrastructureSetup.value = Boolean.FALSE;
						waitForInfrastructureSetup.notifyAll();
					}
					throw new InfrastructureException(
							"Unable to create local Server and to connect to local server",
							serverException).setType(NOT_RUNNING);
				}
			}
		} catch (InfrastructureException e) {
			// defer throw of exception to call
			// waitForInfrastructureSetup.notifyAll()
			// otherwise, possible calls to waitForInfrastructureSetup may
			// wait forever
			exception = e;
			running = false;
		}

		synchronized (waitForInfrastructureSetup) {
			waitForInfrastructureSetup.value = Boolean.FALSE;
			waitForInfrastructureSetup.notifyAll();
		}

		if (exception != null)
			throw exception;
	}

	private boolean sendMessage(Message message) {
		LOGGER.trace("Local: sending " + message);
		return sendMessage_(Tree.fromObject(message));
	}

	/**
	 * @return true if message was sent, false otherwise
	 */
	private boolean sendMessage_(Tree message) {
		synchronized (clientSocket) {
			byte[] b = SExpressionWriter.writeSExpression(message);
			try {
				output.write(b);
				output.flush();
				return true;
			} catch (IOException e) {
				LOGGER.error("Could not send message to local server");
				return false;
			}
		}
	}

	@Override
	public void removeDevice(PublishedDeviceDescription device)
			throws InfrastructureException {
		if (device.isPattern())
			throw new InfrastructureException(
					"Description must not be pattern.").setType(IS_PATTERN);

		waitForInfrastructureSetup();

		synchronized (devices) {
			devices.remove(device);
		}

		sendMessage(Message.removeDevice(device));
	}

	@Override
	public void searchDevice(PublishedDeviceDescription pattern)
			throws InfrastructureException {
		waitForInfrastructureSetup();

		sendMessage(Message.searchDevice(pattern, null));
	}

	@Override
	public PublishedDeviceDescription searchSingleDevice(
			PublishedDeviceDescription pattern, int timeout)
			throws InfrastructureException {
		waitForInfrastructureSetup();

		Object wait = new Object();
		Pair<PublishedDeviceDescription, Object> p = new Pair<PublishedDeviceDescription, Object>(
				pattern, wait);

		synchronized (searches) {
			searches.add(p);
		}

		synchronized (wait) {
			try {
				// call searchDevice here, so wait.notify() is called *after* we
				// call wait.wait();
				searchDevice(pattern);
				wait.wait(timeout);
			} catch (InterruptedException e) {
				// should not happen
			}
		}

		synchronized (searches) {
			if (p.getFirst() == pattern) {// test if the reference is the same
				// if the pattern has not been replaced, remove the query and
				// return null
				searches.remove(p);
				return null;
			}

			// else return the device
			return p.getFirst();
		}
	}

	@Override
	public void addDevice(PublishedDeviceDescription device)
			throws InfrastructureException {
		if (device.isPattern())
			throw new InfrastructureException(
					"Description must not be pattern.").setType(IS_PATTERN);

		waitForInfrastructureSetup();

		synchronized (devices) {
			devices.add(device);
		}

		sendMessage(Message.addDevice(device, false));
	}

	private void parseMessage(Message message) {
		LOGGER.trace("Local: parsing " + message);

		if (message.getType() == REBIND) {
			bindAddress = message.getAddress();
			LOGGER.info("renew binAddress to " + bindAddress);
			return;
		}

		if (message.getType() != EVENT) {
			LOGGER.warn("got something different than an event or rebind, ignore "
					+ message);
			return;
		}

		InfrastructureEvent event = message.getEvent();
		notifyObservers(event);

		InfrastructureEventType eventType = event.getEventType();
		PublishedDeviceDescription device = event.device;

		if (eventType == ANSWER_DEVICE) {
			// check if there are pending "search single device" querys
			synchronized (searches) {
				Iterator<Pair<PublishedDeviceDescription, Object>> it = searches
						.iterator();
				while (it.hasNext()) {
					Pair<PublishedDeviceDescription, Object> p = it.next();
					if (device.matches(p.getFirst())) {
						it.remove();
						// replace pattern with device
						p.setFirst(device);
						synchronized (p.getSecond()) {
							p.getSecond().notifyAll();
						}
					}
				}
			}
		} else if (eventType == SHUT_DOWN_ANNOUNCEMENT) {
			if (event.address.isLoopbackAddress()
					|| event.address.equals(clientSocket.getLocalAddress())) {
				shutDownAnnounceTime = System.currentTimeMillis();
			}
		}
	}

	/**
	 * creates the instance (if not already done) and returns the instance.<br/>
	 * Use {@link #stop()} to stop infrastructure of the current VM.
	 * 
	 * @param bindAddress
	 *            The address to which the Infrastructure ({@link HostServer})
	 *            shoule be bound. If it is <b>null</b>, a random address is
	 *            choosen. This only affects the first Infrastructure, (i.e. if
	 *            a {@link HostServer} is created). If this is not first
	 *            {@link Infrastructure}, {@link #bindAddress} may be changed,
	 *            thus a call to {@link #getBindAddress()} is always
	 *            recommended.
	 * 
	 * @throws InfrastructureException
	 *             when creation of the infrastructure fails.
	 */
	public static DistributedInfrastructure createInstance(
			InetAddress bindAddress) throws InfrastructureException {
		return createInstance(bindAddress, true);
	}

	/**
	 * creates the instance (if not already done) and returns the instance.<br/>
	 * Use {@link #stop()} to stop infrastructure of the current VM.
	 * 
	 * @param bindAddress
	 *            The address to which the Infrastructure ({@link HostServer})
	 *            shoule be bound. If it is <b>null</b>, a random address is
	 *            choosen. This only affects the first Infrastructure, (i.e. if
	 *            a {@link HostServer} is created). If this is not first
	 *            {@link Infrastructure}, {@link #bindAddress} may be changed,
	 *            thus a call to {@link #getBindAddress()} is always
	 *            recommended.
	 * 
	 * @param start
	 *            If the infrastructure should start automatically. If false,
	 *            you have to call {@link #start(InetAddress)} manually.
	 * 
	 * @throws InfrastructureException
	 *             when creation of the infrastructure fails.
	 */
	public static DistributedInfrastructure createInstance(
			InetAddress bindAddress, boolean start)
			throws InfrastructureException {
		synchronized (instanceMonitor) {
			if (instance == null)
				instance = new DistributedInfrastructure(bindAddress, start);
			return instance;
		}
	}

	/**
	 * @return the instance, if it was created with
	 *         {@link #createInstance(InetAddress, boolean)} before -
	 *         <b>null</b> otherwise
	 */
	public static Infrastructure getInstance() {
		return instance;
	}

	@Override
	public void stop() {
		if (!running)
			return;

		running = false;
		// close our connection
		// this will notify others that our devices are removed dirty
		try {
			clientSocket.close();
		} catch (IOException e) {
			// ignore
		}

		// SHUT_DOWN event is fired in InfrastructureThread

		if (hostServer != null) {
			// close all servers after a short wait (to get remove device
			// messages for example)
			try {
				Thread.sleep(250);
			} catch (InterruptedException e) {
				// ignore
			}
			hostServer.stop();
		}
	}

	@Override
	public void addObserver(Observer<InfrastructureEvent> observer) {
		synchronized (observers) {
			observers.add(observer);
		}
	}

	@Override
	public void removeObserver(Observer<InfrastructureEvent> observer) {
		synchronized (observers) {
			observers.remove(observer);
		}
	}

	private void notifyObservers(InfrastructureEvent event) {
		synchronized (observers) {
			for (Observer<InfrastructureEvent> o : observers)
				o.notify(this, event);
		}
	}
}
